/**
 * File: ResultMergeController.java
 * Created by: mhaimel
 * Created on: 23 Jul 2009
 * CVS:  $Id: ResultMergeController.java,v 1.1 2009/08/05 15:16:13 mhaimel Exp $
 */
package uk.ac.ebi.curtain.controller.velvet;

import java.io.File;
import java.io.FileOutputStream;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicReference;

import uk.ac.ebi.curtain.controller.ParallelQueuedController;
import uk.ac.ebi.curtain.controller.SimpleReport;
import uk.ac.ebi.curtain.management.IControllerReport;
import uk.ac.ebi.curtain.processor.velvet.impl.ResultMergeProcessor;
import uk.ac.ebi.curtain.utils.CurtainUncheckedException;
import uk.ac.ebi.curtain.utils.concurrent.BlockingQueueShutdown;
import uk.ac.ebi.curtain.utils.io.impl.FileIO;

/**
 * @author mhaimel
 *
 */
public class ResultMergeController extends ParallelQueuedController<VelvetOutputBean, VelvetOutputBean, ResultMergeProcessor> {

	private final BlockingQueueShutdown<ByteBuffer> queue = new BlockingQueueShutdown<ByteBuffer>();
	private final AtomicReference<Future<Long>> future = new AtomicReference<Future<Long>>();
	private final AtomicReference<MyByteWriter> writer = new AtomicReference<MyByteWriter>(null);
	
	@Override
	protected IControllerReport getReturnCall() {
		return new SimpleReport("Finished");
	}

	@Override
	protected ResultMergeProcessor buildController() {
		MyByteWriter myByteWriter = null;
		while(true){
			myByteWriter = writer.get();
			if(null == myByteWriter){
				File superContigFile = getContext().currentFiles().getSuperContigFile();
				getLog().info("Writing data into " + superContigFile);
				MyByteWriter mwriter = new MyByteWriter(queue, superContigFile, false);
				if(writer.compareAndSet(null, mwriter)){
					Future<Long> process = getProcessorService().process(mwriter);
					getLog().debug("Process submitted " + process);
					future.set(process);
					break;
				}
			} else {
				break;
			}
		}
		ResultMergeProcessor job = getContext().getInjector().getInstance(ResultMergeProcessor.class);
		job.setOutput(queue);
		return job;
	}
	
	@Override
	protected void doShutdown() {
		queue.shutdown();
		try {
			getLog().debug("Retrieve result from future " + future.get());
			if(null != future.get()){
				Future<Long> res = future.getAndSet(null);
				if(null != res){
					Long l = res.get();
				}
			}
		} catch (Exception e) {
			throw new CurtainUncheckedException("Problems while writing Contigs!",e);
		}
		super.doShutdown();
	}
	
	public final class MyByteWriter implements Callable<Long> {
		
		private final BlockingQueueShutdown<ByteBuffer> queue;
		private final File outputFile;
		private final boolean append;
		
		public MyByteWriter(BlockingQueueShutdown<ByteBuffer> queue,File outputFile, boolean append) {
			this.queue = queue;
			this.outputFile = outputFile;
			this.append = append;
		}

		@Override
		public Long call() throws Exception {
			long i = 0; 
			FileChannel out = null;
			try{
				out = new FileOutputStream(outputFile,append).getChannel();
				while(true){
					ByteBuffer buf = queue.take();
					if(null == buf && queue.isShutdown()){
						break;
					}
					int len = buf.array().length;
					int write = out.write(buf);
					if(len > write){
						getLog().error("Problem writing full buffer: "+len+" > " + write+": " + buf.toString());
					}
				}
			} catch (Exception e) {
				throw new CurtainUncheckedException("Problems writing to file " + outputFile,e);
			} finally{
				FileIO.closeQuietly(out);
			}
			return i;
		}
		
	}
	
}
